package com.education.realtime.fuction;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.common.Constant;
import com.education.realtime.util.DimUtil;
import com.education.realtime.util.DruidDSUtil;
import com.education.realtime.util.RedisUtil;
import com.education.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {
    
    private ThreadPoolExecutor threadPool;
    private DruidDataSource dataSource;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        
        dataSource = DruidDSUtil.createDataSource(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
    }
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 获取redis连接和phoenix连接
                Jedis redisClient = null;
                DruidPooledConnection conn = null;
                try {
                    redisClient = RedisUtil.getRedisClient();
                    conn = dataSource.getConnection();
                    
                    // 使用连接读取维度信息
                    JSONObject dim = DimUtil.getDim(redisClient, conn, getTableName(), getDimId(input));
    
                    addDim(input, dim);  // 子类实现这个方法, 需要把维度信息补充的input中
    
                    resultFuture.complete(Collections.singletonList(input));  // 把input放入到后序的流中
                    
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (redisClient != null) {
                        redisClient.close();
                    }
                    if (conn != null) {
                        try {
                            conn.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                }
                
            }
        });
    }
}